package com.litchi.handler;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Pair;
import com.litchi.model.BaseOperationMessage;
import com.litchi.model.DeleteMessage;
import com.litchi.model.InsertMessage;
import com.litchi.model.UpdateMessage;
import io.debezium.data.Envelope;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;

import static io.debezium.data.Envelope.FieldName.*;
import static java.util.stream.Collectors.toMap;

/**
 * @Author: Jiangxx
 * @Date: 2023/10/18
 * @Description:
 */
@Slf4j
@Component
public class ChangeEventHandler implements DebeziumEngine.ChangeConsumer<RecordChangeEvent<SourceRecord>>{

    @Autowired
    private KafkaProducerService kafkaProducerService;

    @Override
    public void handleBatch(List<RecordChangeEvent<SourceRecord>> recordChangeEvents, DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> recordCommitter) throws InterruptedException {
        try {
            for (RecordChangeEvent<SourceRecord> recordChangeEvent : recordChangeEvents) {
                SourceRecord record = recordChangeEvent.record();
                Struct recordStruct = (Struct) record.value();
                if (recordStruct == null) {
                    continue;
                }
                BaseOperationMessage message = buildDataChangeOperationMessage(recordStruct);

                if (!Objects.isNull(message)) {
                    message.setLoaderTime(new Date());
                    kafkaProducerService.sendWithConversion(record.topic(), null, message);
                }
            }
            recordCommitter.markBatchFinished();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    private BaseOperationMessage buildDataChangeOperationMessage(Struct recordStruct) {
        Envelope.Operation operation = Envelope.Operation.forCode((String) recordStruct.get(OPERATION));
        if (operation != Envelope.Operation.READ) {
            Map<String, Object> sourceData = parseChangeDataMap(recordStruct, SOURCE);
            Map<String, Object> beforeData = parseChangeDataMap(recordStruct, BEFORE);
            Map<String, Object> afterData = parseChangeDataMap(recordStruct, AFTER);
            if (operation == Envelope.Operation.CREATE && !CollUtil.isEmpty(afterData)) {
                InsertMessage insertMessage = new InsertMessage();
                insertMessage.setColumnInfo(afterData);
                insertMessage.setOwner(sourceData.get("name").toString());
                insertMessage.setTableName(sourceData.get("table").toString());
                insertMessage.setOpTime(sourceData.get("ts_ms").toString());
                return insertMessage;
            }
            if (operation == Envelope.Operation.UPDATE && !CollUtil.isEmpty(afterData) && !CollUtil.isEmpty(beforeData)) {
                UpdateMessage updateMessage = new UpdateMessage();
                updateMessage.setAfter(afterData);
                updateMessage.setBefore(beforeData);
                updateMessage.setOwner(sourceData.get("name").toString());
                updateMessage.setTableName(sourceData.get("table").toString());
                updateMessage.setOpTime(sourceData.get("ts_ms").toString());
                return updateMessage;
            }
            if (operation == Envelope.Operation.DELETE && !CollUtil.isEmpty(beforeData)) {
                DeleteMessage deleteMessage = new DeleteMessage();
                deleteMessage.setBefore(beforeData);
                deleteMessage.setOwner(sourceData.get("name").toString());
                deleteMessage.setTableName(sourceData.get("table").toString());
                deleteMessage.setOpTime(sourceData.get("ts_ms").toString());
                return deleteMessage;
            }
        }
        return null;
    }

    public Map<String, Object> parseChangeDataMap(Struct sourceRecordStruct, String recordField) {
        Struct struct = (Struct) sourceRecordStruct.get(recordField);
        if (!Objects.isNull(struct)) {
            // 将变更的行封装为Map
            return struct.schema().fields().stream()
                    .map(Field::name)
                    .filter(fieldName -> struct.get(fieldName) != null)
                    .map(fieldName -> Pair.of(fieldName, struct.get(fieldName)))
                    .collect(toMap(Pair::getKey, Pair::getValue));
        }
        return new HashMap<>();
    }


}
